package jk.fox.transport.netty.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.AttributeKey;
import jk.fox.common.RpcDict;
import jk.fox.common.payload.message.Message;
import jk.fox.common.payload.message.MessageDispatcher;
import jk.fox.config.transport.netty.NettyConfiguration;
import jk.fox.transport.ServerTransporter;
import jk.fox.transport.netty.codec.HeartBeatDecode;
import jk.fox.transport.netty.codec.HeartBeatEncode;
import jk.fox.transport.netty.codec.RpcRequestDecode;
import jk.fox.transport.netty.codec.RpcResponseEncode;
import org.apache.log4j.Logger;

import java.net.InetAddress;
import java.net.UnknownHostException;

/**
 * @author cuichao
 * @Description: NettyServerTransporter
 * @Date: create in 2020/4/13 14:45
 */
public class NettyServerTransporter implements ServerTransporter {
    private static Logger logger = Logger.getLogger(NettyServerTransporter.class);

    EventLoopGroup master = new NioEventLoopGroup(1);
    EventLoopGroup worker = new NioEventLoopGroup();


    private MessageDispatcher dispatcher;
    private NettyConfiguration config;

    private ServerBootstrap bootstrap;
    private boolean enableServerHeartBeat;
    private boolean keepAlive;
    private boolean tcp_noDelay;
    private int tcp_backlog;
    private int allIdleTimeSeconds;
    private int readerIdleTimeSeconds;
    private int writerIdleTimeSeconds;


    public NettyServerTransporter(NettyConfiguration config, MessageDispatcher dispatcher){
        super();
        this.dispatcher =dispatcher;
        this.config = config;
        initWithConfig(this.config);
        executor();
    }

    private void initWithConfig(NettyConfiguration config){
        this.enableServerHeartBeat = config.isEnableServerHeartBeat();
        this.keepAlive = config.isKeepAlive();
        this.tcp_backlog = config.getServerTcpBacklog();
        this.tcp_noDelay = config.isDisableNagle();
        this.readerIdleTimeSeconds = config.getReaderIdleTimeSeconds();
        this.writerIdleTimeSeconds = config.getWriterIdleTimeSeconds();
        this.allIdleTimeSeconds = config.getAllIdleTimeSeconds();
    }

    private void executor(){
        bootstrap = new ServerBootstrap();
        RpcServerHandler rpcServerHandler = new RpcServerHandler(dispatcher);

        ChannelInitializer<NioSocketChannel> initializer = new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
                //心跳编解码器
                ch.pipeline().addLast(new HeartBeatEncode());
                ch.pipeline().addLast(new HeartBeatDecode());

                if(enableServerHeartBeat){
                    logger.info(String.format("channel is %s enableHeartBeat,the readerIdle is %d , witterIdle is %d",ch.id().asShortText(), readerIdleTimeSeconds,writerIdleTimeSeconds ));
                    ch.pipeline().addLast(new IdleStateHandler(readerIdleTimeSeconds,writerIdleTimeSeconds,allIdleTimeSeconds));
                }
                //心跳处理器
                ch.pipeline().addLast(new PingHandler());
                //请求响应编解码器
                ch.pipeline().addLast(new RpcResponseEncode());
                ch.pipeline().addLast(new RpcRequestDecode());
                //业务单元执行器
                pipeline.addLast(rpcServerHandler);

            }
        };

        bootstrap
                .group(master,worker)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, tcp_backlog)
                .childHandler(initializer)
                .childAttr(AttributeKey.valueOf(RpcDict.PING_COUNTER),0)
                .handler(new LoggingHandler(LogLevel.DEBUG))
                .childOption(ChannelOption.SO_KEEPALIVE,keepAlive)
                .childOption(ChannelOption.TCP_NODELAY,tcp_noDelay);

    }

    @Override
    public void bind(int port) {
        StringBuilder hostBuilder = new StringBuilder();
        try {
            InetAddress localHost = InetAddress.getLocalHost();
            hostBuilder.append(localHost.getHostAddress());
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
        try {
            String server = hostBuilder.append(":").append(port).toString();
            ChannelFuture bind = bootstrap.bind(port);
            bind.addListener( future -> {
                Message message = new Message();
                message.setStatus(RpcDict.MessageStatus.Init);
                if(future.isSuccess()){
                    //设置启动状态
                    message.payload(RpcDict.SERVER_STATUS,RpcDict.NodeStatus.SUCCESS);
                    message.payload(RpcDict.SERVER_ADDRESS,server);
                    logger.debug(server+" has started");
                }else {
                    //设置启动状态
                    message.payload(RpcDict.SERVER_STATUS,RpcDict.NodeStatus.FAILURE);
                    logger.error(server+" start failure");
                }
                //分发消息状态
                dispatcher.dispatch(message);
            });
            bind.sync();
            bind.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            master.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }
}
